package com.atguigu.flink.datastreamapi.dirstribute;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/1
 *
 *  flink和spark和mr的区别。
 *      spark和mr，多个进程之间传递数据，通过shuffle机制。
 *          shuffle:  MapTask把处理的结果写在本地的磁盘上
 *                   ReduceTask主动去上游的MapTask的磁盘上拉取数据
 *
 *      flink： 上游Task 主动 推送数据 到下游
 *
 *  ----------------------------
 *      如果下游同一个Operator(map操作)有多个并行运行的实例。
 *          上游的Task如何把数据推送到下游指定的 Task？
 *              取决于 ChannelSelector(通道选择器)
 *                       int selectChannel(T record);
 *                              如果是单播，需要一个下游的通道发送
 *                        boolean isBroadcast();
 *                              当前传播方式是否是广播
 *
 *  -------------------------------
 *          当前使用的都是StreamPartitioner的子类进行数据分发。
 *           rescale   RescalePartitioner 。
 *                          在同一个TaskManager内部，保证负载均衡(相对)
 *
 *           reblance   RebalancePartitioner  默认。
 *                          上游按照绝对负载均衡的方式发送到下游。
 *                          大数据只怕数据倾斜
 *
 *           keyBy   KeyGroupStreamPartitioner.
 *                          相同的key一定发送到下游的同一个通道。
 *                          不同的key，可能也会发送到下游的同一个通道。
 *           global   GlobalPartitioner.
 *                      全局汇总。只会发数据汇总发送到下游的 0号 通道。
 *
 *           forward   ForwardPartitioner
 *           shuffle   ShufflePartitioner  随机发送
 *           xxx   CustomPartitionerWrapper
 *
 *           以上的全是单播。 一个数只发到下游的一个通道。
 *           -------------
 *
 *           broadcast   BroadcastPartitioner   广播
 *                       一个数只发到下游的所有通道。
 *                       用于分发配置信息。
 *
 *          可以使用不同的分区器把数据按照某种规则发送到下游。无需自己构造，只需调用flink提供的算子(自己构造分区器)。
 *
 *
 */
public class Demo1_Partition
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);


        /*
                从端口去读数据，Parallelism只能是1

                socketTextStream()本质上：
                        env.addSource( SocketTextStreamFunction x )

                SocketTextStreamFunction implements SourceFunction<String>

                SourceFunction: 非并行运行的Source！
         */
        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888)
                                        // .setParallelism(2) 无效，报错
            ;

        ds
            .forward()
            /*
                强制要求 上下游的 parilisim必须一致

                上游的0号通道，固定把数据发到下游的0号通道
                依次类推
                上游的N号通道，固定把数据发到下游的N号通道
             */
            //.rebalance()
            //.keyBy(s -> s)
             //.rescale()
            .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
